use std::ffi::OsStr;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use fuse3::raw::prelude::*;
use fuse3::{Errno, Result, Timestamp};

use futures_util::stream;
use futures_util::stream::Iter;
use std::iter::Skip;
use std::vec::IntoIter;
use bytes::Bytes;

use tracing::{debug, error, info, trace, warn};

use crate::Scheduler;
use crate::metadata::{NodeAttr, Attribute};

const TTL: Duration = Duration::from_secs(10);

pub struct Mountpoint {
    attr: Attribute,
    user: u32,
    group: u32,
}

impl Mountpoint {
    pub async fn new(attr: Attribute, uid: u32, gid: u32) -> Self {
        Mountpoint {
            attr,
            user: uid,
            group: gid,
        }
    }
}

pub struct VFStorage {
    mountpoint: Mountpoint,
    scheduler: Arc<Scheduler>,
}

impl VFStorage {
    pub async fn new(attr: Attribute, scheduler: Arc<Scheduler>, uid: u32, gid: u32) -> Self {
        let mountpoint = Mountpoint::new(attr, uid, gid).await;
        VFStorage { mountpoint, scheduler }
    }

    fn node_attr(&self, inode: u64, attr: NodeAttr) -> FileAttr {
        match attr {
            NodeAttr::Directory => {
                FileAttr {
                    ino: inode,
                    generation: 0,
                    size: self.mountpoint.attr.size,
                    blocks: 0,
                    atime: Timestamp::new(self.mountpoint.attr.atime, 0),
                    mtime: Timestamp::new(self.mountpoint.attr.mtime, 0),
                    ctime: Timestamp::new(self.mountpoint.attr.ctime, 0),
                    kind: FileType::Directory,
                    perm: 0o664,
                    nlink: 0,
                    uid: self.mountpoint.user,
                    gid: self.mountpoint.group,
                    rdev: self.mountpoint.attr.rdev,
                    blksize: 0,
                }
            }
            NodeAttr::File(_, attribute) => {
                FileAttr {
                    ino: inode,
                    generation: 0,
                    size: attribute.size,
                    blocks: 0,
                    atime: Timestamp::new(attribute.atime, 0),
                    mtime: Timestamp::new(attribute.mtime, 0),
                    ctime: Timestamp::new(attribute.ctime, 0),
                    kind: FileType::RegularFile,
                    perm: 0o664,
                    nlink: 0,
                    uid: self.mountpoint.user,
                    gid: self.mountpoint.group,
                    rdev: attribute.rdev,
                    blksize: 0,
                }
            }
        }
    }

    async fn r#move(&self, parent: u64, name: &OsStr, new_parent: u64, new_name: &OsStr) -> Result<()> {
        self.scheduler.rename_local(parent, name, new_parent, new_name).await.or_else(|e|{
            error!("rename error: {}", e);
            Err(e.into())
        })
    }
}

#[async_trait]
impl Filesystem for VFStorage {
    type DirEntryStream = Iter<Skip<IntoIter<Result<DirectoryEntry>>>>;
    type DirEntryPlusStream = Iter<Skip<IntoIter<Result<DirectoryEntryPlus>>>>;

    async fn init(&self, _req: Request) -> Result<()> {
        Ok(())
    }

    async fn destroy(&self, _req: Request) {}

    async fn lookup(&self, _req: Request, parent: u64, name: &OsStr) -> Result<ReplyEntry> {
        let r = self.scheduler.lookup(parent, name).await;
        match r {
            Some((inode, attr)) => {
                let attr = self.node_attr(inode, attr);
                Ok(ReplyEntry {
                    ttl: TTL,
                    attr,
                    generation: 0,
                })
            }
            None => {
                warn!("lookup error: file not found {}, name {}", parent, name.to_str().unwrap());
                Err(Errno::new_not_exist())
            }
        }
    }

    async fn getattr(&self, _req: Request, inode: u64, _fh: Option<u64>, _flags: u32) -> Result<ReplyAttr> {
        let r = self.scheduler.locate(inode).await;
        match r {
            Some(node) => {
                let attr = self.node_attr(inode, node);
                Ok(ReplyAttr{ttl: TTL, attr})
            },
            None => {
                warn!("get attr error {}", inode);
                Err(Errno::new_not_exist())
            }
        }
    }

    async fn setattr(&self, _req: Request, inode: u64, _fh: Option<u64>, _set_attr: SetAttr) -> Result<ReplyAttr> {
        let r = self.scheduler.locate(inode).await;
        match r {
            Some(node) => {
                let attr = self.node_attr(inode, node);
                Ok(ReplyAttr{ttl: TTL, attr})
            },
            None => {
                debug!("set attr {} error, attr {:?}", inode, _set_attr);
                Err(Errno::new_not_exist())
            }
        }
    }

    /// can not use so far
    async fn mknod(&self, _req: Request, parent: u64, name: &OsStr, _mode: u32, _rdev: u32) -> Result<ReplyEntry> {
        match self.scheduler.create(parent, name).await {
            Err(e) => {
                warn!("create file error: {}", e);
                Err(e.into())
            }
            Ok((node,  _fh, attr)) => {
                let attr = self.node_attr(node, attr);
                Ok(ReplyEntry {
                    ttl: TTL,
                    attr,
                    generation: 0,
                })
            }
        }
    }

    async fn mkdir(&self, _req: Request, parent: u64, name: &OsStr, _mode: u32, _umask: u32, ) -> Result<ReplyEntry> {
        match self.scheduler.mkdir(parent, name).await {
            Ok(id) => {
                let attr = self.node_attr(id, NodeAttr::Directory);
                Ok(ReplyEntry{
                    ttl: TTL,
                    attr,
                    generation: 0,
                })
            }
            Err(e) => {
                error!("mkdir error: {}", e);
                Err(e.into())
            }
        }
    }

    async fn unlink(&self, _req: Request, parent: u64, name: &OsStr) -> Result<()> {
        match self.scheduler.remove_local(parent, name).await {
            Ok(_node) => {
                info!("unlink {}", parent);
                Ok(())
            },
            Err(e) => {
                warn!("unlink error: {}", e);
                Err(e.into())
            }
        }
    }

    async fn rmdir(&self, _req: Request, _parent: u64, _name: &OsStr) -> Result<()> {
        if let Err(e) = self.scheduler.rmdir_local(_parent, _name).await {
            warn!("rmdir error, parent {}, error {}", _parent, e);
            Err(e.into())
        } else {
            Ok(())
        }
    }

    async fn rename(&self, _req: Request, parent: u64, name: &OsStr, new_parent: u64, new_name: &OsStr) -> Result<()> {
        self.r#move(parent, name, new_parent, new_name).await
    }

    async fn open(&self, _req: Request, inode: u64, flags: u32) -> Result<ReplyOpen> {
        let opened = self.scheduler.open_node(inode).await;
        match opened {
            Ok(fh) => Ok(ReplyOpen { fh, flags }),
            Err(_e) => Err(Errno::new_not_exist())
        }
    }

    async fn read(
        &self,
        _req: Request,
        inode: u64,
        fh: u64,
        offset: u64,
        size: u32,
    ) -> Result<ReplyData> {
        match self.scheduler.read(inode, fh, offset, size).await
        {
            Ok(buf) => Ok(ReplyData {
                data: Bytes::copy_from_slice(&buf),
            }),
            Err(_e) => Err(_e.into()),
        }
    }

    async fn write(
        &self,
        _req: Request,
        _inode: u64,
        _fh: u64,
        _offset: u64,
        _data: &[u8],
        _flags: u32,
    ) -> Result<ReplyWrite> {
        match self.scheduler.write(_inode, _fh, _offset, _data).await
        {
            Ok(size) => Ok(ReplyWrite {
                written: size,
            }),
            Err(_e) => Err(_e.into()),
        }
    }

    async fn statfs(&self, _req: Request, _inode: u64) -> Result<ReplyStatFs> {
        match self.scheduler.filesystem().await {
            Ok(f) => Ok(f),
            Err(_e) => Err(Errno::new_not_exist())
        }
    }

    async fn release(
        &self,
        _req: Request,
        _inode: u64,
        _fh: u64,
        _flags: u32,
        _lock_owner: u64,
        _flush: bool,
    ) -> Result<()> {
        match self.scheduler.release(_fh).await {
            Ok(_) => Ok(()),
            Err(_e) => Err(_e.into()),
        }
    }

    async fn fsync(&self, _req: Request, _inode: u64, _fh: u64, _datasync: bool) -> Result<()> {
        Ok(())
    }

    async fn flush(&self, _req: Request, _inode: u64, _fh: u64, _lock_owner: u64) -> Result<()> {
        Ok(())
    }

    async fn opendir(&self, _req: Request, _inode: u64, flags: u32) -> Result<ReplyOpen> {
        Ok(ReplyOpen { fh: 0, flags })
    }

    async fn readdir(
        &self,
        _req: Request,
        inode: u64,
        _fh: u64,
        offset: i64,
    ) -> Result<ReplyDirectory<Self::DirEntryStream>> {
        let vec = self.scheduler.readdir(inode, _fh, offset).await?;
        debug!("readdir, offset {}, length {}", offset, vec.len());

        let entries = vec.iter().enumerate().map(|(index, (id, name, attr))| {
            Ok(DirectoryEntry {
                inode: id.clone(),
                kind: attr.file_type(),
                name: name.to_os_string(),
                offset: (index + 1) as i64,
            })
        }).collect::<Vec<_>>();

        Ok(ReplyDirectory {
            entries: stream::iter(entries.into_iter().skip(offset as usize)),
        })
    }

    async fn access(&self, _req: Request, _inode: u64, _mask: u32) -> Result<()> {
        Ok(())
    }

    async fn create(&self, _req: Request, parent: u64, name: &OsStr, _mode: u32, flags: u32) -> Result<ReplyCreated> {
        match self.scheduler.create(parent, name).await {
            Err(e) => {
                warn!("create file error: {}", e);
                Err(e.into())
            }
            Ok((node, fh, attr)) => {
                let attr = self.node_attr(node, attr);
                Ok(ReplyCreated {
                    ttl: TTL,
                    attr,
                    fh,
                    flags,
                    generation: 0,
                })
            }
        }
    }

    async fn readdirplus(
        &self,
        _req: Request,
        parent: u64,
        _fh: u64,
        offset: u64,
        _lock_owner: u64,
    ) -> Result<ReplyDirectoryPlus<Self::DirEntryPlusStream>> {
        let vec = self.scheduler.readdir(parent, _fh, offset as i64).await?;
        debug!("readdir plus, offset {}, length {}", offset, vec.len());

        let entries = vec.iter().enumerate().map(|(index, (id, name, attr))| {
            let id = id.clone();
            let attr = self.node_attr(id, attr.clone());
            Ok(DirectoryEntryPlus {
                inode: id,
                generation: 0,
                kind: attr.kind,
                name: name.to_os_string(),
                offset: (index + 1) as i64,
                attr,
                entry_ttl: TTL,
                attr_ttl: TTL,
            })
        }).collect::<Vec<_>>();

        Ok(ReplyDirectoryPlus {
            entries: stream::iter(entries.into_iter().skip(offset as usize)),
        })
    }

    async fn rename2(&self, _req: Request, parent: u64, name: &OsStr, new_parent: u64, new_name: &OsStr, _flags: u32) -> Result<()> {
        self.r#move(parent, name, new_parent, new_name).await
    }
}